package com.study.hadoop.io;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

/**
 * HDFS IO操作
 */
public class HadoopIO {

    /**
     * IO方式从本地上传文件到HDFS
     */
    public void putFileToHDFS() throws URISyntaxException, IOException, InterruptedException {
        // 1 获取文件系统
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://localhost:9000"), conf, "stephen");

        // 2 创建输入流
        FileInputStream fis = new FileInputStream(new File("/Users/stephen/Applications/hadoop-2.6.0-cdh5.13.3/README.txt"));

        // 3 获取输出流
        FSDataOutputStream fos = fs.create(new Path("/README.txt"));

        // 4 流对拷
        IOUtils.copyBytes(fis, fos, conf);

        // 5 关闭资源
        IOUtils.closeStream(fos);
        IOUtils.closeStream(fis);
        fs.close();
    }


    /**
     * IO方式从HDFS下载文件到本地
     *
     * @throws IOException
     * @throws URISyntaxException
     * @throws InterruptedException
     */
    public void getFileFromHDFS() throws IOException, URISyntaxException, InterruptedException {
        // 1 获取文件系统
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://localhost:9000"), conf, "stephen");

        // 2 获取输入流
        FSDataInputStream fis = fs.open(new Path("/README.txt"));

        // 3 获取输出流
        FileOutputStream fos = new FileOutputStream(new File("/Users/stephen/README.txt"));

        // 4 流对拷
        IOUtils.copyBytes(fis, fos, conf);

        // 5 关闭资源
        IOUtils.closeStream(fos);
        IOUtils.closeStream(fis);
        fs.close();
    }

    /**
     * 分块读取HDFS文件：读取第一块
     *
     * @throws URISyntaxException
     * @throws IOException
     * @throws InterruptedException
     */
    public void readFileSeekFirstBlock() throws URISyntaxException, IOException, InterruptedException {
        // 1 获取文件系统
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://localhost:9000"), conf, "stephen");

        // 2 打开输入流
        FSDataInputStream fis = fs.open(new Path("/hadoop-2.7.2.tar.gz"));

        // 3 创建输出流
        FileOutputStream fos = new FileOutputStream(new File("/Users/stephen/hadoop-2.7.2.tar.gz.part1"));

        // 4 流拷贝
        byte[] buf = new byte[1024];
        for (int i = 0; i < 1024 * 128; i++) {
            fis.read(buf);
            fos.write(buf);
        }

        // 5 关闭资源
        IOUtils.closeStream(fis);
        IOUtils.closeStream(fos);
        fs.close();

    }

    /**
     * 分块读取HDFS文件：读取第二块至最后
     *
     * @throws URISyntaxException
     * @throws IOException
     * @throws InterruptedException
     */
    public void readFileSeekSecondBlock() throws URISyntaxException, IOException, InterruptedException {
        // 1 获取文件系统
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://localhost:9000"), conf, "stephen");

        // 2 打开输入流
        FSDataInputStream fis = fs.open(new Path("/hadoop-2.7.2.tar.gz"));

        // 3 定位输入数据位置
        fis.seek(1024 * 1024 * 128);

        // 4 创建输出流
        FileOutputStream fos = new FileOutputStream(new File("/Users/stephen/hadoop-2.7.2.tar.gz.part2"));

        // 5 流拷贝
        IOUtils.copyBytes(fis, fos, conf);

        // 6 关闭资源
        IOUtils.closeStream(fis);
        IOUtils.closeStream(fos);
        fs.close();

    }
}
